package com.myrice.jms.impl;

import java.io.IOException;
import java.nio.channels.ByteChannel;
import java.nio.channels.SelectableChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ScheduledFuture;

import com.myrice.core.AccessException;
import com.myrice.core.Connection;
import com.myrice.core.ServerContext;
import com.myrice.core.Session;
import com.myrice.core.SessionFactory;
import com.myrice.core.impl.DefaultConnection;
import com.myrice.core.impl.DefaultMessageReader;
import com.myrice.core.impl.DefaultMessageWriter;
import com.myrice.core.impl.DefaultServerHandler;
import com.myrice.filter.IAcceptFilter;
import com.myrice.filter.IAcceptorFilter;
import com.myrice.filter.IClosedFilter;
import com.myrice.filter.IErrFilter;
import com.myrice.filter.IFilterChain;
import com.myrice.filter.IFilterChain.FilterChain;
import com.myrice.filter.IFilterChain.INextFilter;
import com.myrice.filter.IFilterChain.IPrevFilter;
import com.myrice.filter.IMessageFilter;
import com.myrice.jms.IMsgProtocolFilter;
import com.myrice.jms.Listener;
import com.myrice.jms.MessageFactory;
import com.myrice.jms.MsgFilter;
import com.myrice.jms.MsgFilterChain;
import com.myrice.jms.MsgRequest;
import com.myrice.jms.MsgResponse;
import com.myrice.jms.MsgServerHandler;
import com.myrice.jms.MsgSession;
import com.myrice.jms.SessionListener;

public class DefaultMsgServerHandler extends DefaultServerHandler implements
		MsgServerHandler, IAcceptFilter, IAcceptorFilter, IMessageFilter,
		IClosedFilter, IErrFilter, SessionListener {

	private static final int CACHE_TASK_MAX = 1024;

	private IMsgProtocolFilter protocolFilter;

	private List<SessionListener> sessionListeners;

	private ScheduledFuture<?> checkFuture;

	public DefaultMsgServerHandler(DefaultMsgConnector connector) {
		super(connector);
		sessionListeners = new ArrayList<SessionListener>();
	}

	@Override
	public void init() {
		super.init();

		protocolFilter = instanceProtocolFilter();
		protocolFilter.setMessageFactory(instanceMessageFactory());

		IFilterChain chain = getFilterChain();
		chain.addLastFilter(IFilterChain.FILTER_ACCEPT, this);
		chain.addLastFilter(IFilterChain.FILTER_ACCEPTOR, this);
		chain.addLastFilter(IFilterChain.FILTER_MESSAGE, this);
		chain.addLastFilter(IFilterChain.FILTER_CLOSED, this);
		chain.addLastFilter(IFilterChain.FILTER_ERROR, this);
		chain.addLastFilter(IFilterChain.FILTER_PROTOCOL_ENCODE, protocolFilter);
		chain.addLastFilter(IFilterChain.FILTER_PROTOCOL_DECODE, protocolFilter);

		int length = filters.size();
		for (int i = 0; i < length; i++) {
			filters.get(i).init(this);
		}

		addListener(this);

		for (int i = 0, size = sessionListeners.size(); i < size; i++) {
			sessionListeners.get(i).initializeListener();
		}

		startKeepLiveChecker();

	}

	@Override
	public void destory() {

		stopKeepLiveChecker();

		for (int i = 0, size = sessionListeners.size(); i < size; i++) {
			sessionListeners.get(i).destroyListener();
		}

		removeListener(this);

		IFilterChain chain = getFilterChain();
		chain.removeFilter(IFilterChain.FILTER_ACCEPT, this);
		chain.removeFilter(IFilterChain.FILTER_ACCEPTOR, this);
		chain.removeFilter(IFilterChain.FILTER_MESSAGE, this);
		chain.removeFilter(IFilterChain.FILTER_CLOSED, this);
		chain.removeFilter(IFilterChain.FILTER_ERROR, this);
		chain.removeFilter(IFilterChain.FILTER_PROTOCOL_ENCODE, protocolFilter);
		chain.removeFilter(IFilterChain.FILTER_PROTOCOL_DECODE, protocolFilter);
		protocolFilter.setMessageFactory(null);

		int length = filters.size();
		for (int i = 0; i < length; i++) {
			filters.get(i).destory();
		}
		super.destory();
	}

	private void stopKeepLiveChecker() {
		if (checkFuture != null) {
			checkFuture.cancel(false);
			checkFuture = null;
		}
	}

	/** 注册响应超时检查周期任务 **/
	protected void startKeepLiveChecker() {
		stopKeepLiveChecker();
		checkFuture = schedule(new CheckKeepLiveTask(), 1000, 1000);
	}

	private class CheckKeepLiveTask implements Runnable {

		public void run() {
			long time = System.currentTimeMillis();
			// for (Connection conn : connections.values())
			// ((DefaultMsgSession) conn.getSession()).check4timeout(time);
			List<Session> list = new ArrayList<Session>(getSessionAll());
			Session session = null;
			int size = sessionListeners.size();
			for (int i = 0, count = list.size(); i < count; i++)
				try {
					//				String key = entry.getKey();
					session = list.get(i);
					for (int j = 0; j < size; j++)
						if (!sessionListeners.get(j).onSessionScan(session,
								time))
							break;

				} catch (Throwable e) {
					getNotifier().fireOnError(session, e);
				}
		}
	}

	public DefaultMsgSession getHandle(ByteChannel sc) {
		Connection conn = getConnection(sc);
		if (conn != null)
			return (DefaultMsgSession) conn.getSession();
		return null;
	}

	public void serverAccept(ServerContext serverHandler,
			FilterChain<IAcceptFilter> filterChain) throws Exception {
		if (filterChain.hasNext()) {
			filterChain.nextFilter().serverAccept(serverHandler,
					filterChain.getNext());
		}
	}

	public Connection sessionAccept(ServerContext serverHandler,
			SelectableChannel socket, FilterChain<IAcceptorFilter> filterChain)
			throws Exception {
		Connection conn = null;
		if (filterChain.hasNext()) {
			conn = filterChain.nextFilter().sessionAccept(serverHandler,
					socket, filterChain.getNext());
			if (conn != null
					&& !(conn.getSession() instanceof DefaultMsgSession)) {
				return conn;
			}
		}

		Session session = null;
		if (conn == null) {
			conn = createConnection(socket);
			session = createSession(conn, conn.hashCode());
			conn.setSession(session);// 设置连接的默认Session
		} else {
			session = conn.getSession();
		}

		log.debug(new StringBuilder("Connected: ").append(session)
				.append("  =>  ").append(getConnector()).toString());

		return session.getConnection();
	}

	protected Connection createConnection(SelectableChannel sc) {
		if (sc instanceof ByteChannel) {
			DefaultConnection conn = new DefaultConnection();
			conn.init((ByteChannel) sc);
			return conn;
		}
		throw new IllegalArgumentException("Unsupport Channel:" + sc);
	}

	public Connection sessionOpened(ServerContext serverHandler,
			Connection conn, FilterChain<IAcceptorFilter> filterChain)
			throws Exception {
		if (conn == null)
			return null;

		onSessionOpened(conn.getSession());

		if (filterChain.hasNext()) {
			conn = filterChain.nextFilter().sessionOpened(serverHandler, conn,
					filterChain.getNext());
		}
		return conn;
	}

	@Override
	protected void onSessionOpened(Session session) {
		if (session instanceof DefaultMsgSession) {
			DefaultMsgSession s = (DefaultMsgSession) session;
			s.init(this);
			s.onAccpeted();
		}
		super.onSessionOpened(session);
	}

	public void messageReceived(Session session0, Object message0,
			FilterChain<IMessageFilter> chain) {
		if (session0 instanceof DefaultMsgSession
				&& message0 instanceof DefaultMessage) {
			DefaultMessage message = (DefaultMessage) message0;
			DefaultMsgSession session = message.getSession();
			if (message.isResponse())
				try {
					// ==========响应消息到达========
					session.onResponse(message);
				} catch (Throwable e) {
					getNotifier().fireOnError(session, e);
				} finally {
					message.destory();// 销毁响应消息
				}
			else
				try {
					// ==========请求消息到达========
					takeFilterChain().doFilter(message, message);// 过滤请求消息
				} catch (Throwable e) {
					message.setStatus(MsgResponse.STATUS_INNER_ERROR);// 内部未知错误
					getNotifier().fireOnError(session, e);
				} finally {
					message.flush();// 发送响应消息
					message.destory();// 销毁请求消息
				}
		} else if (chain.hasNext()) {
			chain.nextFilter().messageReceived(session0, message0,
					chain.getNext());
		}
	}

	public void serverExcept(Session session, Throwable e,
			FilterChain<IErrFilter> filterChain) {

		StackTraceElement[] stack = e.getStackTrace();
		if (stack.length < 4
				|| (!stack[stack.length - 4].getClassName().startsWith(
						DefaultMessageReader.class.getCanonicalName()) && !stack[stack.length - 4]
						.getClassName().startsWith(
								DefaultMessageWriter.class.getCanonicalName()))) {
			e.printStackTrace();
		}

		if (filterChain.hasNext()) {
			filterChain.nextFilter().serverExcept(session, e,
					filterChain.getNext());
		}

		if (e instanceof IOException && session != null && session.isDefault()
				&& session.getConnection().isClosed() == false) {
			session.getConnection().close();
		}
	}

	public void sessionClosed(Connection conn,
			FilterChain<IClosedFilter> filterChain) {
		Session session = conn.getSession();

		log.debug("Closed: " + conn.getInetAddress() + "  msg-input: "
				+ session.getMessageInputQueue().size() + "  msg-output: "
				+ session.getMessageOutputQueue().size());

		if (filterChain.hasNext()) {
			filterChain.nextFilter().sessionClosed(conn, filterChain.getNext());
		}

		if (session instanceof DefaultMsgSession) {
			DefaultMsgSession session0 = (DefaultMsgSession) session;
			session0.onClosed();
		}

		if (conn instanceof DefaultConnection)
			((DefaultConnection) conn).onClosed();
	}

	private MsgFilterChain takeFilterChain() {
		com.myrice.jms.MsgFilterChain chain = recycle4filter.poll();
		if (chain == null)
			chain = new FilterChainCache();
		return chain;
	}

	private class FilterChainCache implements com.myrice.jms.MsgFilterChain {
		private int pos;

		public void doFilter(MsgRequest request, MsgResponse response)
				throws Exception {
			if (pos < filters.size()) {
				filters.get(pos++).doFilter(request, response, this);
				pos--;// 回溯
			}
			if (pos == 0) {
				recycle4filter.offer(this);
			}
		}
	}

	@Override
	public MsgSession createSession(Connection conn, Object sessionId) {
		MsgSession session = (MsgSession) super.createSession(conn, sessionId);
		for (int i = 0, size = sessionListeners.size(); i < size; i++) {
			if (!sessionListeners.get(i).onSessionCreate(session))
				throw new AccessException(sessionListeners.get(i)
						+ " SessionListener not allow sreate Session: "
						+ sessionId + "  Connection: " + conn.getInetAddress());
		}
		return session;
	}

	@Override
	public MsgSession removeSession(String sessionId) {
		MsgSession session = getSession(sessionId);
		if (session == null)
			return null;
		for (int i = 0, size = sessionListeners.size(); i < size; i++) {
			Connection conn = session.getConnection();
			if (!sessionListeners.get(i).onSessionDestroy(session))
				throw new AccessException(sessionListeners.get(i)
						+ " SessionListener not allow close Session: "
						+ sessionId + "  Connection: " + conn.getInetAddress());
		}
		super.removeSession(sessionId);
		return session;
	}

	@Override
	public MsgSession getSession(String sessionId) {
		return (MsgSession) super.getSession(sessionId);
	}

	public boolean removeFilter(MsgFilter filter) {
		log.debug("remove filter:" + filter);
		return filters.remove(filter);
	}

	public boolean addLastFilter(MsgFilter filter) {
		log.debug("add last-filter:" + filter);
		return filters.add(filter);
	}

	public void addFirstFilter(MsgFilter filter) {
		log.debug("add first-filter:" + filter);
		filters.add(0, filter);
	}

	/** 过滤器列表 */
	private List<MsgFilter> filters = new ArrayList<MsgFilter>();
	private final Queue<MsgFilterChain> recycle4filter = new ArrayBlockingQueue<MsgFilterChain>(
			CACHE_TASK_MAX);

	public void onAdded(IFilterChain filterChain, IPrevFilter prevFilter) {
	}

	public void onRemoved(IFilterChain filterChain, IPrevFilter prevFilter) {
	}

	public boolean onPrevFilterAdd(IFilterChain filterChain,
			IPrevFilter prevFilter) {
		return true;
	}

	public boolean onNextFilterAdd(IFilterChain filterChain,
			INextFilter nextFilter) {
		return true;
	}

	public boolean onPrevFilterRemove(IFilterChain filterChain,
			IPrevFilter prevFilter) {
		return true;
	}

	public boolean onNextFilterRemove(IFilterChain filterChain,
			INextFilter nextFilter) {
		return true;
	}

	public void initializeListener() {
	}

	public boolean onSessionCreate(Session session) {
		return true;
	}

	public boolean onSessionScan(Session session, long time) {
		//		log.debug("Scan Session:" + session.getSessionId());
		if (session instanceof DefaultMsgSession)
			if (((DefaultMsgSession) session).check4timeout(time))
				try {
					log.debug("Session idle timeout : " + session);
					removeSession(session.getSessionId());
				} catch (Throwable e) {
					getNotifier().fireOnError(session, e);
				}
		return true;
	}

	public boolean onSessionDestroy(Session session) {
		return true;
	}

	public void destroyListener() {
	}

	public MessageFactory instanceMessageFactory() {
		return new DefaultMessageFactory();
	}

	public IMsgProtocolFilter instanceProtocolFilter() {
		return new DefaultMsgProtocolFilter();
	}

	@Override
	public SessionFactory instanceSessionFactory() {
		return new DefaultMsgSessionFactory();
	}

	public IMsgProtocolFilter getProtocolFilter() {
		return protocolFilter;
	}

	public MessageFactory getMessageFactory() {
		return protocolFilter.getMessageFactory();
	}

	public void setMessageFactory(MessageFactory factory) {
		this.protocolFilter.setMessageFactory(factory);
	}

	public boolean addListener(Listener listener) {
		if (listener instanceof SessionListener) {
			return sessionListeners.add((SessionListener) listener);
		}
		return false;
	}

	public boolean removeListener(Listener listener) {
		return sessionListeners.remove(listener);
	}
}
